package org.example.trigger.job;

import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.handler.annotation.XxlJob;
import io.micrometer.core.annotation.Timed;
import lombok.extern.slf4j.Slf4j;
import org.apache.dubbo.common.utils.CollectionUtils;
import org.example.domain.user_raffle.service.IUserRaffleService;
import org.example.types.common.RedisKeysPrefix;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author lxc18
 * @date 2025/3/30 17:13
 * @description UpdatePrizeStockJob
 */
@Slf4j
@Component
public class UpdatePrizeStockJob {
    @Resource
    @Qualifier("threadPoolExecutor") //nacos配置中心也有这个冲突了，需要指定bean的名称
    private ThreadPoolExecutor executor;
    @Resource
    private RedissonClient redissonClient;
    @Resource
    private IUserRaffleService userRaffleService;

    @XxlJob("UpdatePrizeStockJobHandler")
    public ReturnT<String> updatePrizeStockJobHandler() {
        RLock lock = redissonClient.getLock("UpdatePrizeStockJobHandler");

        try {
            boolean isLocked = lock.tryLock(3, 10, TimeUnit.SECONDS);
            if (!isLocked) {
                log.warn("未能获取锁，任务已被其他实例执行");
                return ReturnT.SUCCESS;
            }

            // 获取所有奖品 ID
            List<Long> prizeIds = userRaffleService.getAllPrizeIds();
            if (CollectionUtils.isEmpty(prizeIds)) {
                log.info("没有奖品需要更新库存");
                return ReturnT.SUCCESS;
            }

            // 任务执行
            List<Future<?>> futures = new ArrayList<>();
            for (Long prizeId : prizeIds) {
                Future<?> future = executor.submit(() -> {
                    try {
                        String cacheKey = RedisKeysPrefix.PRIZE_DELAY_QUEUE + prizeId;
                        RBlockingQueue<Object> blockingQueue = redissonClient.getBlockingQueue(cacheKey);
                        // 用一个集合来接收所有的元素
                        List<Object> allItems = new ArrayList<>();
                        blockingQueue.drainTo(allItems);
//
//                        Object poll = blockingQueue.poll(5, TimeUnit.SECONDS);
//                        if (poll == null) {
//                            return;
//                        }
                        // 更新库存
                        userRaffleService.decrPrizeStock(prizeId, allItems.size());
                        log.info("定时任务，更新库存到数据库：{}", allItems.size());
                    } catch (Exception e) {
                        log.error("奖品 {} 库存更新失败", prizeId, e);
                    }
                });
                futures.add(future);
            }

            // 等待所有任务完成
            for (Future<?> future : futures) {
                try {
                    future.get();
                } catch (Exception e) {
                    log.error("异步任务执行失败", e);
                }
            }

        } catch (Exception e) {
            log.error("定时任务更新奖品库存失败", e);
        } finally {
            if (lock.isLocked() && lock.isHeldByCurrentThread()) {
                lock.unlock();
            }
        }

        return ReturnT.SUCCESS;
    }

}
